package com.hexinfo.dmpro.sparing.service.impl;


import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import cn.hutool.core.util.StrUtil;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.dto.DateDataDTO;
import com.hexinfo.dmpro.sparing.dto.DropDateDataDTO;
import com.hexinfo.dmpro.sparing.dto.HiveDefaultPartDTO;
import com.hexinfo.dmpro.sparing.service.AsyncDateDataService;
import com.hexinfo.dmpro.sparing.service.HotStandbyLogTableService;
import com.hexinfo.dmpro.sparing.util.Logger;
import com.xqfunds.job.core.log.XxlJobLogger;
import liquibase.pro.packaged.S;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.io.File;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

@Service
@Slf4j
public class AsyncDateDataServiceImpl implements AsyncDateDataService {

    @Value("${heatBackup.logPath}")
    private String logPath;          //每张表日志打印路径
    @Value("${hdfs.tarPrefix}")
    private String tarPrefix;      //同步删除，目标库前缀
    @Value("${hdfs.hiveDefaultPartition}")
    private String dev;             //hive 默认分区

    @Autowired
    @Qualifier("connPoolImpalaSource")
    private DataSource connPoolImpalaSource;
    @Autowired
    @Qualifier("connPoolImpalaTarget")
    private DataSource connPoolImpalaTarget;

    @Autowired
    private HotStandbyLogTableService hotStandbyLogTableService;

    private String dropSql = "alter table {} drop if exists partition ({} cast({} as int) IN ({}))";
    private String dropDevSql = "alter table {} drop if exists partition (cast({} as string) = '{}')";
    private String souCol = "cast({} as string) = '{}' ,";
    private String soColSql = "SELECT {} FROM {} where 1=1 GROUP BY {}";

    //表完成数量
    private Map<String, Integer> tableSizeMap = new ConcurrentHashMap<String, Integer>();
    private synchronized void addTableSize(String souName) {
        tableSizeMap.put(souName, tableSizeMap.getOrDefault(souName,0)+1);
    }
    public synchronized Integer getTableSize(String souName) {
        return tableSizeMap.get(souName);
    }
    public synchronized void deleteTableSize(String souName) {
        tableSizeMap.remove(souName);
    }

    //结束标志
    private Map<String, Boolean> endFlagMap = new ConcurrentHashMap<String, Boolean>();
    public synchronized void updateEndFlag(String souName,Boolean state) {
        endFlagMap.put(souName, state);
    }
    public synchronized Boolean getEndFlag(String souName) {
        return endFlagMap.get(souName);
    }
    public synchronized void deleteEndFlag(String souName) {
        endFlagMap.remove(souName);
    }

    //记录任务执行是否有异常
    private Map<String, Boolean> taskStateMap = new ConcurrentHashMap<String, Boolean>();
    public synchronized void updateTaskState(String souName,Boolean state) {
        taskStateMap.put(souName, state);
    }
    public synchronized Boolean getTaskState(String souName) {
        return taskStateMap.get(souName);
    }
    public synchronized void deleteTaskState(String souName) {
        taskStateMap.remove(souName);
    }

    @Override
    @Async("threadPoolTaskExecutor")
    public Future<Boolean> execSql(DateDataDTO dateDataDTO, String source, int sizes, FileAppender heatlog, String tbLogPath) {
        if (StrUtil.isNotBlank(dateDataDTO.getTableId())){
            //表级日志分析记录修改--用时详情修改
            hotStandbyLogTableService.upHotLog(dateDataDTO.getTableId(),"【同步删除】"+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"--");
        }
        //表日志文件缓存
        String strPath = tbLogPath + "/" + dateDataDTO.getDbTbName() + ".txt";
        FileAppender tableLog = new FileAppender(new File(strPath), 100, true);
        tableLog.append(logTime("-------------------------【同步删除开始】"));
        //----------------------------操作方法--------------------------------------
        String str = dateTable(dateDataDTO,source,heatlog,tableLog);
        //----------------------------判断表是不是全部完成----------------------------
//        addTableSize(source);    //表完成数量加一
//        Integer integer = getTableSize(source);
        logLog(heatlog,str,"1");
        tableLog.append(logTime(str));
        tableLog.append(logTime("-------------------------【同步删除结束】"));
        tableLog.flush();
        logLog(heatlog,"","2"); //刷进日志中
        if (StrUtil.isNotBlank(dateDataDTO.getTableId())){
            //表级日志分析记录修改--用时详情修改
            hotStandbyLogTableService.upHotLog(dateDataDTO.getTableId(),DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"\n");
        }
//        if (integer.equals(sizes) || integer == sizes){
//            //----------------------------所有表全部执行完成----------------------------
//            deleteTableSize(source);  //移除key
//            updateEndFlag(source,true);
//        }
        return new AsyncResult<>(true);
    }

    /**
     * 获取、拼接、删除、刷新
     */
    private String dateTable(DateDataDTO dateDataDTO,String source,FileAppender heatlog, FileAppender tableLog) {
        try {
            // ① 删除两年前数据
//            Connection tarConnDropTwo = connPoolImpalaTarget.getConnection();
//            Statement tarStTwo = tarConnDropTwo.createStatement();
//            execDropSql(tarConnDropTwo,tarStTwo, dateDataDTO.getDelSql(), heatlog, tableLog,source);

            //获取日期分区数据--执行状态
            Map<String,Boolean> map = new HashMap<>();
            map.put("state",true);
            // ② 获取查询每张表的日期分区
            Connection souConn = connPoolImpalaSource.getConnection();
            Set<String> souDateValues = selectDateTable(souConn,dateDataDTO.getSouSql(),dateDataDTO.getDateColName(),
                    heatlog, tableLog,source,dateDataDTO.getTableId(),map);
            //关闭目标连接
            closeConn(souConn,heatlog,tableLog,source);
            if (!map.get("state")){
                return tarPrefix+dateDataDTO.getDbTbName()+"查询【源】日期分区执行命令【异常】";
            }
            Connection tarConn = connPoolImpalaTarget.getConnection();
            Set<String> tarDateValues = selectDateTable(tarConn,dateDataDTO.getTarSql(),dateDataDTO.getDateColName(),
                    heatlog, tableLog,source,dateDataDTO.getTableId(),map);
            //关闭目标连接
            closeConn(tarConn,heatlog,tableLog,source);
            if (!map.get("state")){
                return tarPrefix+dateDataDTO.getDbTbName()+"查询【目标】日期分区执行命令【异常】";
            }
            // ③ 获取dropSql
            String dropSql = getDropSql(souDateValues, tarDateValues, dateDataDTO, source);
            if(StrUtil.isNotBlank(dropSql)){
                extracted(dateDataDTO, source, heatlog, tableLog, dropSql);
                return "【"+tarPrefix+dateDataDTO.getDbTbName()+"删除目标分区完成】";
            }else{
                return tarPrefix+dateDataDTO.getDbTbName()+"的目标表时间分区没有比源多的时间分区，不做删除操作";
            }
            //采用查询hive元数据的方式来进行删除信息
//            if(StrUtil.isNotBlank(dateDataDTO.getSouColName())&&StrUtil.isNotBlank(dateDataDTO.getDbTbName())){
//                try{
//                    //获取查询每张表的源分区
//                    String soSql = StrUtil.format(soColSql, dateDataDTO.getSouColName(),tarPrefix+dateDataDTO.getDbTbName(), dateDataDTO.getSouColName());
//                    log.warn("========获取查询每张表的源分区:"+soSql);
//                    Connection souSoConn = connPoolImpalaSource.getConnection();
//                    Set<String> souSoValues = selectSoureTable(souSoConn,soSql,dateDataDTO.getSouColName(),
//                            heatlog, tableLog,source,dateDataDTO.getTableId(),map);
//                    //关闭目标连接
//                    closeConn(souSoConn,heatlog,tableLog,source);
//                    if (!map.get("state")){
//                        return tarPrefix+dateDataDTO.getDbTbName()+"查询【源】源分区执行命令【异常】";
//                    }
//                    Connection tarSoConn = connPoolImpalaTarget.getConnection();
//                    Set<String> tarSoValues = selectSoureTable(tarSoConn,soSql,dateDataDTO.getSouColName(),
//                            heatlog, tableLog,source,dateDataDTO.getTableId(),map);
//                    //关闭目标连接
//                    closeConn(tarSoConn,heatlog,tableLog,source);
//                    if (!map.get("state")){
//                        return tarPrefix+dateDataDTO.getDbTbName()+"查询【目标】源分区执行命令【异常】";
//                    }
//                    // 获取dropSql 默认分区
//                    HiveDefaultPartDTO dto = dropHiveDefaultPartition(souDateValues,tarDateValues,souSoValues,tarSoValues,dateDataDTO);
//                    if(StrUtil.isNotBlank(dto.getSourceSql())){
//                        extracted(dateDataDTO, source, heatlog, tableLog, dto.getSourceSql());
//                    }
//                    if(StrUtil.isNotBlank(dto.getDateSql())){
//                        extracted(dateDataDTO, source, heatlog, tableLog, dto.getDateSql());
//                    }
//                    if(StrUtil.isNotBlank(dto.getDateSql())||StrUtil.isNotBlank(dto.getSourceSql())){
//                        return "【"+tarPrefix+dateDataDTO.getDbTbName()+"删除目标分区完成】"+"\n【"+tarPrefix+dateDataDTO.getDbTbName()+"删除目标默认分区完成】";
//                    }else if(StrUtil.isNotBlank(dropSql)){
//                        return "【"+tarPrefix+dateDataDTO.getDbTbName()+"删除目标分区完成】";
//                    }else{
//                        return tarPrefix+dateDataDTO.getDbTbName()+"的目标表时间分区没有比源多的时间分区，不做删除操作";
//                    }
//                }catch (Exception e){
//                    log.warn("=============="+tarPrefix+dateDataDTO.getDbTbName()+"同步删除默认分区SQL",e.getMessage(),e);
//                }
//            }
//            return "【"+tarPrefix+dateDataDTO.getDbTbName()+"删除目标分区完成】";
        } catch (Exception e) {
            if (StrUtil.isNotBlank(dateDataDTO.getTableId())){
                //表级日志分析记录修改--执行状态修改
                hotStandbyLogTableService.upState(dateDataDTO.getTableId(), ScanCommonConstants.ClusterName.SCSB.value);
            }
            updateTaskState(source,false);    //修改任务异常状态
            return tarPrefix+dateDataDTO.getDbTbName()+"查询日期分区获取连接池连接【异常】："+e;
        }
    }

    private void extracted(DateDataDTO dateDataDTO, String source, FileAppender heatlog, FileAppender tableLog, String dropSql) throws SQLException {
        if (StrUtil.isNotBlank(dropSql)){
            log.warn("===============同步删除分区SQL"+dropSql);
            Connection tarConn1 = connPoolImpalaTarget.getConnection();
            // ④ 执行dropSql
            execDropSql(tarConn1, dropSql, heatlog, tableLog, source, dateDataDTO.getTableId());
            //关闭目标连接
            closeConn(tarConn1,heatlog,tableLog,source);
        }
    }


    /**
     * 执行dropSql
     */
    private void execDropSql(Connection conn,String sql,FileAppender heatlog, FileAppender tableLog,
                             String source,String tableId) {
        try{
            @Cleanup
            Statement st = conn.createStatement();
            int i = st.executeUpdate(sql);
            tableLog.append(logTime("执行dropSql："+sql+"--执行状态："+i+"（成功）"));
        }catch (Exception e){
            logLog(heatlog,"执行dropSql："+sql+"【异常】："+e,"1");
            tableLog.append(logTime("执行dropSql："+sql+"【异常】："+e));
//            Logger.log("执行dropSql："+sql+"【异常】："+e);
            if (StrUtil.isNotBlank(tableId)){
                //表级日志分析记录修改--执行状态修改
                hotStandbyLogTableService.upState(tableId, ScanCommonConstants.ClusterName.SCSB.value);
            }
            updateTaskState(source,false);    //修改任务异常状态
        }
//        finally {
//            //关闭目标连接
//            closeConn(conn,heatlog,tableLog,source);
//        }
    }

    /**
     * 查询每张表的日期分区
     */
    private Set<String> selectDateTable(Connection conn,String sql,String dateColName,
                                        FileAppender heatlog, FileAppender tableLog,String source,String tableId,Map<String,Boolean> map) {
        Set<String> dateValues = new HashSet<>();
        try{
            @Cleanup
            Statement souSt = conn.createStatement();
            @Cleanup
            ResultSet resultSet = souSt.executeQuery(sql);
            tableLog.append(logTime("查询日期分区："+sql));
            //获取日期分区
            while (resultSet.next()) {
                dateValues.add(resultSet.getString(dateColName));
            }
        }catch (Exception e){
            logLog(heatlog,"查询日期分区："+sql+"【异常】："+e,"1");
            tableLog.append(logTime("查询日期分区："+sql+"【异常】："+e));
//            Logger.log("查询日期分区："+sql+"【异常】："+e);
            if (StrUtil.isNotBlank(tableId)){
                //表级日志分析记录修改--执行状态修改
                hotStandbyLogTableService.upState(tableId, ScanCommonConstants.ClusterName.SCSB.value);
            }
            updateTaskState(source,false);    //修改任务异常状态
            map.put("state",false);
        }
//        finally {
//            //关闭连接
//            closeConn(conn,heatlog,tableLog,source);
//        }
        return dateValues;
    }


    /**
     * 查询每张表的源分区
     */
    private Set<String> selectSoureTable(Connection conn,String sql,String souColName,
                                        FileAppender heatlog, FileAppender tableLog,String source,String tableId,Map<String,Boolean> map) {
        Set<String> soureValues = new HashSet<>();
        try{
            @Cleanup
            Statement st = conn.createStatement();
            @Cleanup
            ResultSet resultSet = st.executeQuery(sql);
            tableLog.append(logTime("查询源分区："+sql));
            log.warn("========查询源分区:"+sql);
            //获取日期分区
            while (resultSet.next()) {
                soureValues.add(resultSet.getString(souColName));
            }
        }catch (Exception e){
            logLog(heatlog,"查询源分区："+sql+"【异常】："+e,"1");
            tableLog.append(logTime("查询源分区："+sql+"【异常】："+e));
            log.warn("========查询源分区异常");
            if (StrUtil.isNotBlank(tableId)){
                //表级日志分析记录修改--执行状态修改
                hotStandbyLogTableService.upState(tableId, ScanCommonConstants.ClusterName.SCSB.value);
            }
            updateTaskState(source,false);    //修改任务异常状态
            map.put("state",false);
        }
//        finally {
//            //关闭连接
//            closeConn(conn,heatlog,tableLog,source);
//        }
        return soureValues;
    }

    /**
     * 关闭连接
     */
    private void closeConn(Connection conn,FileAppender heatlog, FileAppender tableLog,String source) {
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (Exception e) {
            logLog(heatlog,"同步删除关闭连接【异常】："+e,"1");
            tableLog.append(logTime("同步删除关闭连接【异常】："+e));
//            Logger.log("同步删除关闭连接【异常】："+e);
            updateTaskState(source,false);    //修改任务异常状态
        }
    }

    /**
     * @Method dropHiveDefaultPartition
     * @Param souDateValues
     * @Param tarDateValues
     * @Param dateDataDTO
     * @Param souCol
     * @Param souColSql
     * @Return java.lang.String
     * @Description 删除默认分区处理
     * @Author Wei MinYe
     * @Date 2024/3/26 16:46
     * @Version V1.0
     */
    private HiveDefaultPartDTO dropHiveDefaultPartition(Set<String> souDateValues, Set<String> tarDateValues,
                                                        Set<String> souSoValues,Set<String> tarSoValues,DateDataDTO dateDataDTO){
        HiveDefaultPartDTO dto = new HiveDefaultPartDTO();
        //删除时间默认分区
        dto.setDateSql(extracted(souDateValues, tarDateValues, dateDataDTO));
        //删除源默认分区
        dto.setSourceSql(extracted(souSoValues, tarSoValues, dateDataDTO));
        return dto;
    }

    private String extracted(Set<String> souValues, Set<String> tarValues, DateDataDTO dateDataDTO) {
        String sql = "";
        if(CollectionUtil.isNotEmpty(tarValues)){
            if(CollectionUtil.isNotEmpty(souValues)){
                //获取目标中比源多的日期分区
                tarValues.removeAll(souValues);
            }
            //是否包含默认分区
            if(tarValues.contains(dev)){
                //判断是否存在非日期目录不一致的文件
                sql = StrUtil.format(dropDevSql, tarPrefix+ dateDataDTO.getDbTbName(), dateDataDTO.getDateColName(),dev);
            }
        }
        if(StrUtil.isNotBlank(sql)){
            log.warn("========删除默认分区SQL"+sql);
        }
        return sql;
    }

    /**
     * 获取dropSql
     */
    private String getDropSql(Set<String> souDateValues, Set<String> tarDateValues,
                              DateDataDTO dateDataDTO, String source) {
        String tarDropSql = "";
        //判断目标中是否有分区值
        if(CollectionUtil.isNotEmpty(souDateValues) && CollectionUtil.isNotEmpty(tarDateValues)){
            //获取目标中比源多的日期分区
            tarDateValues.removeAll(souDateValues);
            //判断过滤后是否为空
            if (CollectionUtil.isNotEmpty(tarDateValues)){
                tarDropSql = splitDropSql(tarDateValues,dateDataDTO,source);
            }
        }else if (CollectionUtil.isEmpty(souDateValues) && CollectionUtil.isNotEmpty(tarDateValues)){
            //源分区为空，目标时间分区全删
            tarDropSql = splitDropSql(tarDateValues,dateDataDTO,source);
        }
        return tarDropSql;
    }

    /**
     * 拼接dropSql
     */
    private String splitDropSql(Set<String> tarDateValues, DateDataDTO dateDataDTO, String source) {
        //  "alter table ods.acct_tprofitcurrent drop if exists partition (cast(srcsys as string) = 'HSTA4' , cast(d_cdate as int) IN (20211127,20211128,.....))"
        //拼接in中的参数
        String dates = "";
        for (String tarDateValue : tarDateValues) {
            if(StrUtil.isNotBlank(tarDateValue)){
                if(!"NULL".equals(tarDateValue.toUpperCase())){
                    dates += tarDateValue+",";
                }
            }
        }
        String substring = dates.substring(0, dates.length()-1);
        String tarDropSql = "";
        //拼接dropSql,判断有没有源分区
        if (StrUtil.isNotBlank(dateDataDTO.getSouColName())) {
            String souColSql = StrUtil.format(souCol, dateDataDTO.getSouColName(), dateDataDTO.getSource());
            tarDropSql = StrUtil.format(dropSql, tarPrefix+dateDataDTO.getDbTbName(), souColSql,dateDataDTO.getDateColName(),substring);
        }else {
            tarDropSql = StrUtil.format(dropSql, tarPrefix+dateDataDTO.getDbTbName(), "",dateDataDTO.getDateColName(),substring);
        }
        return tarDropSql;
    }

    /**
     * 加锁，防止并发问题
     */
    private synchronized void logLog(FileAppender heatlog, String logStr ,String str) {
        if (str.equals("1")){
            heatlog.append(logTime(logStr));
        }else {
            heatlog.flush();     //刷进日志中
        }
    }

    /**
     * 日志加日期时间
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
